{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Copyright (c) 2020, NVIDIA CORPORATION.\n",
    "Licensed under the Apache License, Version 2.0 (the \"License\");\n",
    "you may not use this file except in compliance with the License.\n",
    "You may obtain a copy of the License at\n",
    "    http://www.apache.org/licenses/LICENSE-2.0\n",
    "Unless required by applicable law or agreed to in writing, software\n",
    "distributed under the License is distributed on an \"AS IS\" BASIS,\n",
    "WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
    "See the License for the specific language governing permissions and\n",
    "limitations under the License."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ~~Tweet_id is encoded using approximate hashing which caused hashing collisions. Exact encoding of tweet_id is blocked by a bug in cudf. All other columns are using exact encoding.~~ \n",
    "Fixed"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os, time\n",
    "os.environ[\"CUDA_VISIBLE_DEVICES\"]=\"0,1,2,3\"\n",
    "start = time.time()\n",
    "very_start = time.time()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'0.14.0'"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#import pandas as pd, \n",
    "import numpy as np\n",
    "from datetime import datetime\n",
    "import matplotlib.pyplot as plt\n",
    "#pd.set_option('display.max_columns', 500)\n",
    "#pd.set_option('display.max_rows', 500)\n",
    "import cudf, cupy, time, rmm\n",
    "cudf.__version__"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask as dask, dask_cudf\n",
    "from dask.distributed import Client, wait\n",
    "from dask_cuda import LocalCUDACluster\n",
    "import subprocess"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table style=\"border: 2px solid white;\">\n",
       "<tr>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3 style=\"text-align: left;\">Client</h3>\n",
       "<ul style=\"text-align: left; list-style: none; margin: 0; padding: 0;\">\n",
       "  <li><b>Scheduler: </b>ucx://10.2.61.36:38271</li>\n",
       "  <li><b>Dashboard: </b><a href='http://10.2.61.36:8787/status' target='_blank'>http://10.2.61.36:8787/status</a></li>\n",
       "</ul>\n",
       "</td>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3 style=\"text-align: left;\">Cluster</h3>\n",
       "<ul style=\"text-align: left; list-style:none; margin: 0; padding: 0;\">\n",
       "  <li><b>Workers: </b>4</li>\n",
       "  <li><b>Cores: </b>4</li>\n",
       "  <li><b>Memory: </b>270.39 GB</li>\n",
       "</ul>\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "<Client: 'ucx://10.2.61.36:38271' processes=4 threads=4, memory=270.39 GB>"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cluster = LocalCUDACluster(ip='10.2.61.36',protocol=\"ucx\", \n",
    "                           rmm_pool_size=\"31GB\",\n",
    "                           enable_tcp_over_ucx=True, enable_nvlink=True)\n",
    "#cluster = LocalCUDACluster()\n",
    "client = Client(cluster)\n",
    "client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "number of rows: 121386431\n",
      "CPU times: user 2.73 s, sys: 513 ms, total: 3.25 s\n",
      "Wall time: 12.4 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "NP = 16\n",
    "path = '/raid/data/recsys'\n",
    "df = dask_cudf.read_csv(f'{path}/input/training.tsv', sep='\\x01', header=None)#, dtype=DTYPES)\n",
    "df = df.repartition(npartitions=NP)\n",
    "df, = dask.persist(df)\n",
    "_ = wait(df)\n",
    "print('number of rows:',len(df))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 416 ms, sys: 39.2 ms, total: 455 ms\n",
      "Wall time: 2.85 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "features = [\n",
    "    'text_tokens',    ###############\n",
    "    'hashtags',       #Tweet Features\n",
    "    'tweet_id',       #\n",
    "    'media',          #\n",
    "    'links',          #\n",
    "    'domains',        #\n",
    "    'tweet_type',     #\n",
    "    'language',       #\n",
    "    'timestamp',      ###############\n",
    "    'a_user_id',              ###########################\n",
    "    'a_follower_count',       #Engaged With User Features\n",
    "    'a_following_count',      #\n",
    "    'a_is_verified',          #\n",
    "    'a_account_creation',     ###########################\n",
    "    'b_user_id',              #######################\n",
    "    'b_follower_count',       #Engaging User Features\n",
    "    'b_following_count',      #\n",
    "    'b_is_verified',          #\n",
    "    'b_account_creation',     #######################\n",
    "    'b_follows_a',    #################### Engagement Features\n",
    "    'reply',          #Target Reply\n",
    "    'retweet',        #Target Retweet    \n",
    "    'retweet_comment',#Target Retweet with comment\n",
    "    'like',           #Target Like\n",
    "                      ####################\n",
    "]\n",
    "df.columns = features\n",
    "\n",
    "df = df.drop('text_tokens', axis=1)\n",
    "df, = dask.persist(df)\n",
    "_ = wait(df)\n",
    "#df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "hashtags               object\n",
       "tweet_id               object\n",
       "media                  object\n",
       "links                  object\n",
       "domains                object\n",
       "tweet_type             object\n",
       "language               object\n",
       "timestamp               int64\n",
       "a_user_id              object\n",
       "a_follower_count        int64\n",
       "a_following_count       int64\n",
       "a_is_verified            bool\n",
       "a_account_creation      int64\n",
       "b_user_id              object\n",
       "b_follower_count        int64\n",
       "b_following_count       int64\n",
       "b_is_verified            bool\n",
       "b_account_creation      int64\n",
       "b_follows_a              bool\n",
       "reply                 float64\n",
       "retweet               float64\n",
       "retweet_comment       float64\n",
       "like                  float64\n",
       "dtype: object"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 2.76 s, sys: 3.54 s, total: 6.3 s\n",
      "Wall time: 3.47 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df['id']   = 1\n",
    "df['id']   = df['id'].cumsum()\n",
    "df['id'] = df['id'].astype('int32')\n",
    "\n",
    "df['reply']   = df['reply'].fillna(0)\n",
    "df['retweet'] = df['retweet'].fillna(0)\n",
    "df['retweet_comment'] = df['retweet_comment'].fillna(0)\n",
    "df['like']    = df['like'].fillna(0)\n",
    "\n",
    "df['reply']   = df['reply'].astype('int32')\n",
    "df['retweet'] = df['retweet'].astype('int32')\n",
    "df['retweet_comment'] = df['retweet_comment'].astype('int32')\n",
    "df['like']    = df['like'].astype('int32')\n",
    "df, = dask.persist(df)\n",
    "_ = wait(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 708 ms, sys: 58 ms, total: 766 ms\n",
      "Wall time: 2.08 s\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>hashtags</th>\n",
       "      <th>tweet_id</th>\n",
       "      <th>media</th>\n",
       "      <th>links</th>\n",
       "      <th>domains</th>\n",
       "      <th>tweet_type</th>\n",
       "      <th>language</th>\n",
       "      <th>timestamp</th>\n",
       "      <th>a_user_id</th>\n",
       "      <th>a_follower_count</th>\n",
       "      <th>...</th>\n",
       "      <th>b_follower_count</th>\n",
       "      <th>b_following_count</th>\n",
       "      <th>b_is_verified</th>\n",
       "      <th>b_account_creation</th>\n",
       "      <th>b_follows_a</th>\n",
       "      <th>reply</th>\n",
       "      <th>retweet</th>\n",
       "      <th>retweet_comment</th>\n",
       "      <th>like</th>\n",
       "      <th>id</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>None</td>\n",
       "      <td>E7D6C5094767223F6F8789A87A1937AB</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>TopLevel</td>\n",
       "      <td>22C448FF81263D4BAF2A176145EE9EAD</td>\n",
       "      <td>1581262691</td>\n",
       "      <td>D557B03872EF8986F7F4426AE094B2FE</td>\n",
       "      <td>986</td>\n",
       "      <td>...</td>\n",
       "      <td>94</td>\n",
       "      <td>648</td>\n",
       "      <td>False</td>\n",
       "      <td>1478011810</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>83D6C79F5FCEC8D1CAD9E82C2C261611\\tFFAD2DCF664C...</td>\n",
       "      <td>129F4A868712BA2B98D31AF98C3066E4</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>Retweet</td>\n",
       "      <td>22C448FF81263D4BAF2A176145EE9EAD</td>\n",
       "      <td>1581497241</td>\n",
       "      <td>424822AC982CE0E8965506C63B44EC12</td>\n",
       "      <td>1225</td>\n",
       "      <td>...</td>\n",
       "      <td>1139</td>\n",
       "      <td>46</td>\n",
       "      <td>False</td>\n",
       "      <td>1540395738</td>\n",
       "      <td>True</td>\n",
       "      <td>0</td>\n",
       "      <td>1581497559</td>\n",
       "      <td>0</td>\n",
       "      <td>1581497622</td>\n",
       "      <td>2</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>None</td>\n",
       "      <td>04C6C2175852CDBBC23B2446C7E7C22D</td>\n",
       "      <td>None</td>\n",
       "      <td>DDFFB4C01DB85921C3580F614575AA6D</td>\n",
       "      <td>BE4539C53C53FFABCFD232DB100C792B</td>\n",
       "      <td>TopLevel</td>\n",
       "      <td>22C448FF81263D4BAF2A176145EE9EAD</td>\n",
       "      <td>1580978528</td>\n",
       "      <td>1EC14E26417AA926095530AC591BA9CE</td>\n",
       "      <td>3016</td>\n",
       "      <td>...</td>\n",
       "      <td>780</td>\n",
       "      <td>440</td>\n",
       "      <td>False</td>\n",
       "      <td>1432084055</td>\n",
       "      <td>True</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1581060554</td>\n",
       "      <td>3</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>None</td>\n",
       "      <td>168157826315514C120494D4DF8E6216</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>Retweet</td>\n",
       "      <td>D3164C7FBCF2565DDF915B1B3AEFB1DC</td>\n",
       "      <td>1581321849</td>\n",
       "      <td>9B9595B6FEB8948BDDF0D222F27E0118</td>\n",
       "      <td>2121</td>\n",
       "      <td>...</td>\n",
       "      <td>1</td>\n",
       "      <td>45</td>\n",
       "      <td>False</td>\n",
       "      <td>1534313747</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1581328518</td>\n",
       "      <td>4</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>None</td>\n",
       "      <td>B3E3673782A69D9D8A45D3B222F0B073</td>\n",
       "      <td>Photo</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>TopLevel</td>\n",
       "      <td>22C448FF81263D4BAF2A176145EE9EAD</td>\n",
       "      <td>1580956787</td>\n",
       "      <td>525DC99B7CB8F1AC4AD3E66C53FA38E0</td>\n",
       "      <td>813505</td>\n",
       "      <td>...</td>\n",
       "      <td>171</td>\n",
       "      <td>388</td>\n",
       "      <td>False</td>\n",
       "      <td>1490166885</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1580957807</td>\n",
       "      <td>5</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 24 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "                                            hashtags  \\\n",
       "0                                               None   \n",
       "1  83D6C79F5FCEC8D1CAD9E82C2C261611\\tFFAD2DCF664C...   \n",
       "2                                               None   \n",
       "3                                               None   \n",
       "4                                               None   \n",
       "\n",
       "                           tweet_id  media                             links  \\\n",
       "0  E7D6C5094767223F6F8789A87A1937AB   None                              None   \n",
       "1  129F4A868712BA2B98D31AF98C3066E4   None                              None   \n",
       "2  04C6C2175852CDBBC23B2446C7E7C22D   None  DDFFB4C01DB85921C3580F614575AA6D   \n",
       "3  168157826315514C120494D4DF8E6216   None                              None   \n",
       "4  B3E3673782A69D9D8A45D3B222F0B073  Photo                              None   \n",
       "\n",
       "                            domains tweet_type  \\\n",
       "0                              None   TopLevel   \n",
       "1                              None    Retweet   \n",
       "2  BE4539C53C53FFABCFD232DB100C792B   TopLevel   \n",
       "3                              None    Retweet   \n",
       "4                              None   TopLevel   \n",
       "\n",
       "                           language   timestamp  \\\n",
       "0  22C448FF81263D4BAF2A176145EE9EAD  1581262691   \n",
       "1  22C448FF81263D4BAF2A176145EE9EAD  1581497241   \n",
       "2  22C448FF81263D4BAF2A176145EE9EAD  1580978528   \n",
       "3  D3164C7FBCF2565DDF915B1B3AEFB1DC  1581321849   \n",
       "4  22C448FF81263D4BAF2A176145EE9EAD  1580956787   \n",
       "\n",
       "                          a_user_id  a_follower_count  ...  b_follower_count  \\\n",
       "0  D557B03872EF8986F7F4426AE094B2FE               986  ...                94   \n",
       "1  424822AC982CE0E8965506C63B44EC12              1225  ...              1139   \n",
       "2  1EC14E26417AA926095530AC591BA9CE              3016  ...               780   \n",
       "3  9B9595B6FEB8948BDDF0D222F27E0118              2121  ...                 1   \n",
       "4  525DC99B7CB8F1AC4AD3E66C53FA38E0            813505  ...               171   \n",
       "\n",
       "   b_following_count  b_is_verified  b_account_creation  b_follows_a  reply  \\\n",
       "0                648          False          1478011810        False      0   \n",
       "1                 46          False          1540395738         True      0   \n",
       "2                440          False          1432084055         True      0   \n",
       "3                 45          False          1534313747        False      0   \n",
       "4                388          False          1490166885        False      0   \n",
       "\n",
       "      retweet  retweet_comment        like  id  \n",
       "0           0                0           0   1  \n",
       "1  1581497559                0  1581497622   2  \n",
       "2           0                0  1581060554   3  \n",
       "3           0                0  1581328518   4  \n",
       "4           0                0  1580957807   5  \n",
       "\n",
       "[5 rows x 24 columns]"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "df['timestamp']         = df['timestamp'].astype( np.int32 )\n",
    "df['a_follower_count']  = df['a_follower_count'].astype( np.int32 )\n",
    "df['a_following_count'] = df['a_following_count'].astype( np.int32 )\n",
    "df['a_account_creation']= df['a_account_creation'].astype( np.int32 )\n",
    "df['b_follower_count']  = df['b_follower_count'].astype( np.int32 )\n",
    "df['b_following_count'] = df['b_following_count'].astype( np.int32 )\n",
    "df['b_account_creation']= df['b_account_creation'].astype( np.int32 )\n",
    "\n",
    "df, = dask.persist(df)\n",
    "_ = wait(df)\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "hashtags              object\n",
       "tweet_id              object\n",
       "media                 object\n",
       "links                 object\n",
       "domains               object\n",
       "tweet_type            object\n",
       "language              object\n",
       "timestamp              int32\n",
       "a_user_id             object\n",
       "a_follower_count       int32\n",
       "a_following_count      int32\n",
       "a_is_verified           bool\n",
       "a_account_creation     int32\n",
       "b_user_id             object\n",
       "b_follower_count       int32\n",
       "b_following_count      int32\n",
       "b_is_verified           bool\n",
       "b_account_creation     int32\n",
       "b_follows_a             bool\n",
       "reply                  int32\n",
       "retweet                int32\n",
       "retweet_comment        int32\n",
       "like                   int32\n",
       "id                     int32\n",
       "dtype: object"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "number of rows: 12434735\n",
      "CPU times: user 309 ms, sys: 49.9 ms, total: 359 ms\n",
      "Wall time: 1.26 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "dv = dask_cudf.read_csv(f'{path}/input/val.tsv', sep='\\x01', header=None)\n",
    "dv = dv.repartition(npartitions=NP)\n",
    "dv, = dask.persist(dv)\n",
    "_ = wait(dv)\n",
    "print('number of rows:',len(dv))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 126 ms, sys: 8.06 ms, total: 134 ms\n",
      "Wall time: 166 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "features = [\n",
    "    'text_tokens',    ###############\n",
    "    'hashtags',       #Tweet Features\n",
    "    'tweet_id',       #\n",
    "    'media',          #\n",
    "    'links',          #\n",
    "    'domains',        #\n",
    "    'tweet_type',     #\n",
    "    'language',       #\n",
    "    'timestamp',      ###############\n",
    "    'a_user_id',              ###########################\n",
    "    'a_follower_count',       #Engaged With User Features\n",
    "    'a_following_count',      #\n",
    "    'a_is_verified',          #\n",
    "    'a_account_creation',     ###########################\n",
    "    'b_user_id',              #######################\n",
    "    'b_follower_count',       #Engaging User Features\n",
    "    'b_following_count',      #\n",
    "    'b_is_verified',          #\n",
    "    'b_account_creation',     #######################\n",
    "    'b_follows_a',    #################### Engagement Features\n",
    "    #'reply',          #Target Reply\n",
    "    #'retweet',        #Target Retweet    \n",
    "    #'retweet_comment',#Target Retweet with comment\n",
    "    #'like',           #Target Like\n",
    "                      ####################\n",
    "]\n",
    "dv.columns = features\n",
    "dv = dv.drop('text_tokens', axis=1)\n",
    "dv, = dask.persist(dv)\n",
    "_ = wait(dv)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.28 s, sys: 54.6 ms, total: 1.33 s\n",
      "Wall time: 1.69 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "dv['reply']           = 0\n",
    "dv['retweet']         = 0\n",
    "dv['retweet_comment'] = 0\n",
    "dv['like']            = 0\n",
    "\n",
    "dv['id']   = 1\n",
    "dv['id']   = dv['id'].cumsum()\n",
    "dv['id'] = dv['id'] + len(df)\n",
    "dv['id'] = dv['id'].astype('int32')\n",
    "\n",
    "dv['reply']           = dv['reply'].astype( np.int32 )\n",
    "dv['retweet']         = dv['retweet'].astype( np.int32 )\n",
    "dv['retweet_comment'] = dv['retweet_comment'].astype( np.int32 )\n",
    "dv['like']            = dv['like'].astype( np.int32 )\n",
    "\n",
    "dv['timestamp']         = dv['timestamp'].astype( np.int32 )\n",
    "dv['a_follower_count']  = dv['a_follower_count'].astype( np.int32 )\n",
    "dv['a_following_count'] = dv['a_following_count'].astype( np.int32 )\n",
    "dv['a_account_creation']= dv['a_account_creation'].astype( np.int32 )\n",
    "dv['b_follower_count']  = dv['b_follower_count'].astype( np.int32 )\n",
    "dv['b_following_count'] = dv['b_following_count'].astype( np.int32 )\n",
    "dv['b_account_creation']= dv['b_account_creation'].astype( np.int32 )\n",
    "dv, = dask.persist(dv)\n",
    "_ = wait(dv)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 223 ms, sys: 29.9 ms, total: 252 ms\n",
      "Wall time: 1.19 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "dt = dask_cudf.read_csv(f'{path}/input/competition_test.tsv', sep='\\x01', header=None)\n",
    "dt = dt.repartition(npartitions=NP)\n",
    "dt, = dask.persist(dt)\n",
    "_ = wait(dt)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "number of rows: 12434838\n",
      "CPU times: user 164 ms, sys: 9.26 ms, total: 173 ms\n",
      "Wall time: 224 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "dt.columns = features\n",
    "dt = dt.drop('text_tokens', axis=1)\n",
    "dt, = dask.persist(dt)\n",
    "_ = wait(dt)\n",
    "print('number of rows:',len(dt))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(Delayed('int-5a2f7cf3-929e-499d-a027-b14667a91d84'), 24) (Delayed('int-8adc3760-4900-4e29-8570-6569817bb49c'), 24) (Delayed('int-d3a42868-850b-44bd-92b0-2b9dac4d9342'), 24)\n",
      "CPU times: user 1.41 s, sys: 57.5 ms, total: 1.47 s\n",
      "Wall time: 2.26 s\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>hashtags</th>\n",
       "      <th>tweet_id</th>\n",
       "      <th>media</th>\n",
       "      <th>links</th>\n",
       "      <th>domains</th>\n",
       "      <th>tweet_type</th>\n",
       "      <th>language</th>\n",
       "      <th>timestamp</th>\n",
       "      <th>a_user_id</th>\n",
       "      <th>a_follower_count</th>\n",
       "      <th>...</th>\n",
       "      <th>b_follower_count</th>\n",
       "      <th>b_following_count</th>\n",
       "      <th>b_is_verified</th>\n",
       "      <th>b_account_creation</th>\n",
       "      <th>b_follows_a</th>\n",
       "      <th>reply</th>\n",
       "      <th>retweet</th>\n",
       "      <th>retweet_comment</th>\n",
       "      <th>like</th>\n",
       "      <th>id</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>None</td>\n",
       "      <td>04746004AA1F5498834CE7A4C6343D1A</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>TopLevel</td>\n",
       "      <td>22C448FF81263D4BAF2A176145EE9EAD</td>\n",
       "      <td>1581759640</td>\n",
       "      <td>6720CC7830F94CB7465CA283300DB010</td>\n",
       "      <td>119</td>\n",
       "      <td>...</td>\n",
       "      <td>111</td>\n",
       "      <td>673</td>\n",
       "      <td>False</td>\n",
       "      <td>1478011810</td>\n",
       "      <td>True</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>133821167</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>024FE90EC2C01B3CDC46A5A90D66B020\\t1B78BDD9C7FF...</td>\n",
       "      <td>B5C4CBE185831F3E5A58A4D81118D4C7</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>Retweet</td>\n",
       "      <td>22C448FF81263D4BAF2A176145EE9EAD</td>\n",
       "      <td>1581668217</td>\n",
       "      <td>7DDC67265CFB6E0B4820E0BD0E33A8D3</td>\n",
       "      <td>189</td>\n",
       "      <td>...</td>\n",
       "      <td>111</td>\n",
       "      <td>673</td>\n",
       "      <td>False</td>\n",
       "      <td>1478011810</td>\n",
       "      <td>True</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>133821168</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>None</td>\n",
       "      <td>6B6836351BFAA6D1CC1EB0386BCB8C6A</td>\n",
       "      <td>Video</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>Retweet</td>\n",
       "      <td>D3164C7FBCF2565DDF915B1B3AEFB1DC</td>\n",
       "      <td>1582046459</td>\n",
       "      <td>5456A10C7E4F7A415948EA88BE6845D6</td>\n",
       "      <td>4312</td>\n",
       "      <td>...</td>\n",
       "      <td>1150</td>\n",
       "      <td>48</td>\n",
       "      <td>False</td>\n",
       "      <td>1540395738</td>\n",
       "      <td>True</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>133821169</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2D09C59493DAC82D8054E79343DFE76A</td>\n",
       "      <td>0DCF558E40500F22F84F98C4E7C38EDC</td>\n",
       "      <td>Photo</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>Retweet</td>\n",
       "      <td>125C57F4FA6D4E110983FB11B52EFD4E</td>\n",
       "      <td>1582083666</td>\n",
       "      <td>9D421C234C7B59A0EDC8D85C847D4569</td>\n",
       "      <td>272</td>\n",
       "      <td>...</td>\n",
       "      <td>781</td>\n",
       "      <td>442</td>\n",
       "      <td>False</td>\n",
       "      <td>1432084055</td>\n",
       "      <td>True</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>133821170</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>A0AD2EB95B7C918A311D9432E9A8FF7A</td>\n",
       "      <td>F13AA57F12DD6107D9D8544A27BDE9EC</td>\n",
       "      <td>Photo</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>Retweet</td>\n",
       "      <td>D3164C7FBCF2565DDF915B1B3AEFB1DC</td>\n",
       "      <td>1581779241</td>\n",
       "      <td>F63ECD1C7827E767E7C44E9A717056AC</td>\n",
       "      <td>1020</td>\n",
       "      <td>...</td>\n",
       "      <td>15</td>\n",
       "      <td>123</td>\n",
       "      <td>False</td>\n",
       "      <td>1385502405</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>133821171</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 24 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "                                            hashtags  \\\n",
       "0                                               None   \n",
       "1  024FE90EC2C01B3CDC46A5A90D66B020\\t1B78BDD9C7FF...   \n",
       "2                                               None   \n",
       "3                   2D09C59493DAC82D8054E79343DFE76A   \n",
       "4                   A0AD2EB95B7C918A311D9432E9A8FF7A   \n",
       "\n",
       "                           tweet_id  media links domains tweet_type  \\\n",
       "0  04746004AA1F5498834CE7A4C6343D1A   None  None    None   TopLevel   \n",
       "1  B5C4CBE185831F3E5A58A4D81118D4C7   None  None    None    Retweet   \n",
       "2  6B6836351BFAA6D1CC1EB0386BCB8C6A  Video  None    None    Retweet   \n",
       "3  0DCF558E40500F22F84F98C4E7C38EDC  Photo  None    None    Retweet   \n",
       "4  F13AA57F12DD6107D9D8544A27BDE9EC  Photo  None    None    Retweet   \n",
       "\n",
       "                           language   timestamp  \\\n",
       "0  22C448FF81263D4BAF2A176145EE9EAD  1581759640   \n",
       "1  22C448FF81263D4BAF2A176145EE9EAD  1581668217   \n",
       "2  D3164C7FBCF2565DDF915B1B3AEFB1DC  1582046459   \n",
       "3  125C57F4FA6D4E110983FB11B52EFD4E  1582083666   \n",
       "4  D3164C7FBCF2565DDF915B1B3AEFB1DC  1581779241   \n",
       "\n",
       "                          a_user_id  a_follower_count  ...  b_follower_count  \\\n",
       "0  6720CC7830F94CB7465CA283300DB010               119  ...               111   \n",
       "1  7DDC67265CFB6E0B4820E0BD0E33A8D3               189  ...               111   \n",
       "2  5456A10C7E4F7A415948EA88BE6845D6              4312  ...              1150   \n",
       "3  9D421C234C7B59A0EDC8D85C847D4569               272  ...               781   \n",
       "4  F63ECD1C7827E767E7C44E9A717056AC              1020  ...                15   \n",
       "\n",
       "   b_following_count  b_is_verified  b_account_creation  b_follows_a  reply  \\\n",
       "0                673          False          1478011810         True      0   \n",
       "1                673          False          1478011810         True      0   \n",
       "2                 48          False          1540395738         True      0   \n",
       "3                442          False          1432084055         True      0   \n",
       "4                123          False          1385502405        False      0   \n",
       "\n",
       "   retweet  retweet_comment  like         id  \n",
       "0        0                0     0  133821167  \n",
       "1        0                0     0  133821168  \n",
       "2        0                0     0  133821169  \n",
       "3        0                0     0  133821170  \n",
       "4        0                0     0  133821171  \n",
       "\n",
       "[5 rows x 24 columns]"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "dt['reply']           = 0\n",
    "dt['retweet']         = 0\n",
    "dt['retweet_comment'] = 0\n",
    "dt['like']            = 0\n",
    "\n",
    "dt['id']   = 1\n",
    "dt['id']   = dt['id'].cumsum()\n",
    "dt['id']   = dt['id']+len(df)+len(dv)\n",
    "dt['id']   = dt['id'].astype('int32')\n",
    "\n",
    "dt['reply']           = dt['reply'].astype( np.int32 )\n",
    "dt['retweet']         = dt['retweet'].astype( np.int32 )\n",
    "dt['retweet_comment'] = dt['retweet_comment'].astype( np.int32 )\n",
    "dt['like']            = dt['like'].astype( np.int32 )\n",
    "\n",
    "dt['timestamp']         = dt['timestamp'].astype( np.int32 )\n",
    "dt['a_follower_count']  = dt['a_follower_count'].astype( np.int32 )\n",
    "dt['a_following_count'] = dt['a_following_count'].astype( np.int32 )\n",
    "dt['a_account_creation']= dt['a_account_creation'].astype( np.int32 )\n",
    "dt['b_follower_count']  = dt['b_follower_count'].astype( np.int32 )\n",
    "dt['b_following_count'] = dt['b_following_count'].astype( np.int32 )\n",
    "dt['b_account_creation']= dt['b_account_creation'].astype( np.int32 )\n",
    "\n",
    "dt, = dask.persist(dt)\n",
    "_ = wait(dt)\n",
    "print(df.shape,dv.shape,dt.shape)\n",
    "dt.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "121386431 12434735 12434838\n"
     ]
    }
   ],
   "source": [
    "train_size = len(df)#.shape[0]\n",
    "test0_size = len(dv)#.shape[0]\n",
    "test1_size = len(dt)#.shape[0]\n",
    "print(train_size,test0_size,test1_size)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 180 ms, sys: 3.85 ms, total: 184 ms\n",
      "Wall time: 178 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df = dask_cudf.concat( [df,dv,dt] )\n",
    "df, = dask.persist(df)\n",
    "wait(df)\n",
    "del dv, dt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0    22C448FF81263D4BAF2A176145EE9EAD\n",
       "1    22C448FF81263D4BAF2A176145EE9EAD\n",
       "2    22C448FF81263D4BAF2A176145EE9EAD\n",
       "3    D3164C7FBCF2565DDF915B1B3AEFB1DC\n",
       "4    22C448FF81263D4BAF2A176145EE9EAD\n",
       "Name: language, dtype: object"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head()['language']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'dask_cudf.core.DataFrame'>\n",
      "Columns: 24 entries, hashtags to id\n",
      "dtypes: object(9), bool(3), int32(12)"
     ]
    }
   ],
   "source": [
    "df.info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "48 146256004\n"
     ]
    }
   ],
   "source": [
    "print(df.npartitions,len(df))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "\"\\n%%time\\ndf['tweet_id'] = df['tweet_id'].map_partitions(lambda cudf:cudf.hash_encode(stop=1_000_000_000))\\ndf['tweet_id'] = df['tweet_id'].astype( np.int32 )\\n#df['tweet_id'] = df['tweet_id'].map_partitions(lambda cudf:cudf.hash_values()%1_000_000_000)\\ndf, = dask.persist(df)\\n_ = wait(df)\\ndf.head()\\n\""
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "\"\"\"\n",
    "%%time\n",
    "df['tweet_id'] = df['tweet_id'].map_partitions(lambda cudf:cudf.hash_encode(stop=1_000_000_000))\n",
    "df['tweet_id'] = df['tweet_id'].astype( np.int32 )\n",
    "#df['tweet_id'] = df['tweet_id'].map_partitions(lambda cudf:cudf.hash_values()%1_000_000_000)\n",
    "df, = dask.persist(df)\n",
    "_ = wait(df)\n",
    "df.head()\n",
    "\"\"\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 516 ms, sys: 34.1 ms, total: 551 ms\n",
      "Wall time: 929 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df['media'] = df['media'].fillna( '' )\n",
    "def split_join(ds,sep):\n",
    "    df = ds.str.split(sep)\n",
    "    df[0] = df[0].fillna('')\n",
    "    df[1] = df[1].fillna('')\n",
    "    res = df[0]+'_'+df[1]\n",
    "    del df\n",
    "    return res\n",
    "\n",
    "df['media'] = df['media'].map_partitions( lambda x:  split_join(x,'\\t'), meta=('O'))\n",
    "\n",
    "df, = dask.persist(df)\n",
    "_ = wait(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "def factorize_small_cardinality(df,col):\n",
    "    tmp_col = f'{col}_encode'\n",
    "    tmp = df[col].unique().compute()\n",
    "    tmp = tmp.to_frame().reset_index()\n",
    "    tmp = tmp.rename(mapper={'index':tmp_col})\n",
    "    df = df.merge(tmp,on=col,how='left')\n",
    "    df, = dask.persist(df)\n",
    "    wait(df)\n",
    "    head=df.head()\n",
    "    del tmp\n",
    "    df = df.drop(col,axis=1)\n",
    "    df, = dask.persist(df)\n",
    "    wait(df)\n",
    "    df.columns = [i if i!=tmp_col else col for i in df.columns ]\n",
    "    return df,head"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 2.94 s, sys: 210 ms, total: 3.15 s\n",
      "Wall time: 5.71 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "for col in ['language','tweet_type','media']:\n",
    "    df,_ = factorize_small_cardinality(df,col)\n",
    "    df[col] = df[col].astype('int8')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>hashtags</th>\n",
       "      <th>tweet_id</th>\n",
       "      <th>links</th>\n",
       "      <th>domains</th>\n",
       "      <th>timestamp</th>\n",
       "      <th>a_user_id</th>\n",
       "      <th>a_follower_count</th>\n",
       "      <th>a_following_count</th>\n",
       "      <th>a_is_verified</th>\n",
       "      <th>a_account_creation</th>\n",
       "      <th>...</th>\n",
       "      <th>b_account_creation</th>\n",
       "      <th>b_follows_a</th>\n",
       "      <th>reply</th>\n",
       "      <th>retweet</th>\n",
       "      <th>retweet_comment</th>\n",
       "      <th>like</th>\n",
       "      <th>id</th>\n",
       "      <th>language</th>\n",
       "      <th>tweet_type</th>\n",
       "      <th>media</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>None</td>\n",
       "      <td>0A679DD46B17082ECB54EB8D98258FB8</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1580986075</td>\n",
       "      <td>0133906BF22993CDEE417D41BB5A0FAC</td>\n",
       "      <td>9993</td>\n",
       "      <td>8570</td>\n",
       "      <td>False</td>\n",
       "      <td>1481293579</td>\n",
       "      <td>...</td>\n",
       "      <td>1525806228</td>\n",
       "      <td>True</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>18593</td>\n",
       "      <td>63</td>\n",
       "      <td>0</td>\n",
       "      <td>12</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>None</td>\n",
       "      <td>F1CDA01EC4815A2D974DB296A609059A</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1581474056</td>\n",
       "      <td>E3A820CAF4B0044C737C8ED8C4F57828</td>\n",
       "      <td>31707</td>\n",
       "      <td>30367</td>\n",
       "      <td>False</td>\n",
       "      <td>1249459733</td>\n",
       "      <td>...</td>\n",
       "      <td>1486151643</td>\n",
       "      <td>True</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>18594</td>\n",
       "      <td>54</td>\n",
       "      <td>1</td>\n",
       "      <td>12</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>6012776AECA5D0EC9EBF4B8D72FF555A</td>\n",
       "      <td>E15954621FFA57271636473E104C2088</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1581198029</td>\n",
       "      <td>84C77F8066337B94F3CA3DD07E2C0F3A</td>\n",
       "      <td>2582264</td>\n",
       "      <td>1043</td>\n",
       "      <td>True</td>\n",
       "      <td>1288129925</td>\n",
       "      <td>...</td>\n",
       "      <td>1557711616</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>18595</td>\n",
       "      <td>54</td>\n",
       "      <td>2</td>\n",
       "      <td>4</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>None</td>\n",
       "      <td>2D944D13F92A3652BF3D431D28F9B931</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1580967091</td>\n",
       "      <td>15ED5A140B41D20C1E6FF654FB02591F</td>\n",
       "      <td>251</td>\n",
       "      <td>418</td>\n",
       "      <td>False</td>\n",
       "      <td>1443844675</td>\n",
       "      <td>...</td>\n",
       "      <td>1540838549</td>\n",
       "      <td>True</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1580981213</td>\n",
       "      <td>18596</td>\n",
       "      <td>54</td>\n",
       "      <td>0</td>\n",
       "      <td>12</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>None</td>\n",
       "      <td>8BE2354E4F03DCF2F34B7A1917254531</td>\n",
       "      <td>ADCE39195E117002CD6DD0C2B5C0C31A</td>\n",
       "      <td>BBA1D3359A23689C6783652EA73B12BA</td>\n",
       "      <td>1581536516</td>\n",
       "      <td>5480E8F28BFC097044990E17C91A6292</td>\n",
       "      <td>510762</td>\n",
       "      <td>2118</td>\n",
       "      <td>True</td>\n",
       "      <td>1262371833</td>\n",
       "      <td>...</td>\n",
       "      <td>1534792958</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>18597</td>\n",
       "      <td>59</td>\n",
       "      <td>2</td>\n",
       "      <td>4</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 24 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "                           hashtags                          tweet_id  \\\n",
       "0                              None  0A679DD46B17082ECB54EB8D98258FB8   \n",
       "1                              None  F1CDA01EC4815A2D974DB296A609059A   \n",
       "2  6012776AECA5D0EC9EBF4B8D72FF555A  E15954621FFA57271636473E104C2088   \n",
       "3                              None  2D944D13F92A3652BF3D431D28F9B931   \n",
       "4                              None  8BE2354E4F03DCF2F34B7A1917254531   \n",
       "\n",
       "                              links                           domains  \\\n",
       "0                              None                              None   \n",
       "1                              None                              None   \n",
       "2                              None                              None   \n",
       "3                              None                              None   \n",
       "4  ADCE39195E117002CD6DD0C2B5C0C31A  BBA1D3359A23689C6783652EA73B12BA   \n",
       "\n",
       "    timestamp                         a_user_id  a_follower_count  \\\n",
       "0  1580986075  0133906BF22993CDEE417D41BB5A0FAC              9993   \n",
       "1  1581474056  E3A820CAF4B0044C737C8ED8C4F57828             31707   \n",
       "2  1581198029  84C77F8066337B94F3CA3DD07E2C0F3A           2582264   \n",
       "3  1580967091  15ED5A140B41D20C1E6FF654FB02591F               251   \n",
       "4  1581536516  5480E8F28BFC097044990E17C91A6292            510762   \n",
       "\n",
       "   a_following_count  a_is_verified  a_account_creation  ...  \\\n",
       "0               8570          False          1481293579  ...   \n",
       "1              30367          False          1249459733  ...   \n",
       "2               1043           True          1288129925  ...   \n",
       "3                418          False          1443844675  ...   \n",
       "4               2118           True          1262371833  ...   \n",
       "\n",
       "  b_account_creation  b_follows_a  reply  retweet  retweet_comment  \\\n",
       "0         1525806228         True      0        0                0   \n",
       "1         1486151643         True      0        0                0   \n",
       "2         1557711616        False      0        0                0   \n",
       "3         1540838549         True      0        0                0   \n",
       "4         1534792958        False      0        0                0   \n",
       "\n",
       "         like     id  language  tweet_type  media  \n",
       "0           0  18593        63           0     12  \n",
       "1           0  18594        54           1     12  \n",
       "2           0  18595        54           2      4  \n",
       "3  1580981213  18596        54           0     12  \n",
       "4           0  18597        59           2      4  \n",
       "\n",
       "[5 rows x 24 columns]"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 2.08 s, sys: 161 ms, total: 2.25 s\n",
      "Wall time: 6.38 s\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>tweet_id</th>\n",
       "      <th>tweet_encode</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>1109137</th>\n",
       "      <td>0000012429A02D1B5C871FBA53A0C4DD</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2999311</th>\n",
       "      <td>0000043B5500353E778A6B78498EE7CD</td>\n",
       "      <td>2</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7068428</th>\n",
       "      <td>000004B4208284C156C06BCFAB500ACC</td>\n",
       "      <td>3</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4666149</th>\n",
       "      <td>00000BF2119CD5F74998D3D407F15DB8</td>\n",
       "      <td>4</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>107413</th>\n",
       "      <td>00000D200443B9FE776CA28D56ECAD3C</td>\n",
       "      <td>5</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                 tweet_id  tweet_encode\n",
       "1109137  0000012429A02D1B5C871FBA53A0C4DD             1\n",
       "2999311  0000043B5500353E778A6B78498EE7CD             2\n",
       "7068428  000004B4208284C156C06BCFAB500ACC             3\n",
       "4666149  00000BF2119CD5F74998D3D407F15DB8             4\n",
       "107413   00000D200443B9FE776CA28D56ECAD3C             5"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "tweet = df[['tweet_id']]\n",
    "tweet = tweet.drop_duplicates(split_out=16)\n",
    "tweet['tweet_encode'] = 1\n",
    "tweet['tweet_encode'] = tweet['tweet_encode'].cumsum()\n",
    "tweet, = dask.persist(tweet)\n",
    "_ = wait(tweet)\n",
    "tweet.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 6.45 s, sys: 517 ms, total: 6.97 s\n",
      "Wall time: 27.9 s\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>hashtags</th>\n",
       "      <th>links</th>\n",
       "      <th>domains</th>\n",
       "      <th>timestamp</th>\n",
       "      <th>a_user_id</th>\n",
       "      <th>a_follower_count</th>\n",
       "      <th>a_following_count</th>\n",
       "      <th>a_is_verified</th>\n",
       "      <th>a_account_creation</th>\n",
       "      <th>b_user_id</th>\n",
       "      <th>...</th>\n",
       "      <th>b_follows_a</th>\n",
       "      <th>reply</th>\n",
       "      <th>retweet</th>\n",
       "      <th>retweet_comment</th>\n",
       "      <th>like</th>\n",
       "      <th>id</th>\n",
       "      <th>language</th>\n",
       "      <th>tweet_type</th>\n",
       "      <th>media</th>\n",
       "      <th>tweet_id</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1581525142</td>\n",
       "      <td>46019E3576BD47D19BC83E3A8B04EC35</td>\n",
       "      <td>7717</td>\n",
       "      <td>7688</td>\n",
       "      <td>False</td>\n",
       "      <td>1245066162</td>\n",
       "      <td>4992CD23C7A65F9D6CB1F6067D94D09C</td>\n",
       "      <td>...</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1516532</td>\n",
       "      <td>54</td>\n",
       "      <td>1</td>\n",
       "      <td>12</td>\n",
       "      <td>2725610</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1581252945</td>\n",
       "      <td>185A653D674A8BDA4C351617C84E5311</td>\n",
       "      <td>781</td>\n",
       "      <td>498</td>\n",
       "      <td>False</td>\n",
       "      <td>1476364476</td>\n",
       "      <td>5FAA46A494872C5210B49C5CFCFAD331</td>\n",
       "      <td>...</td>\n",
       "      <td>True</td>\n",
       "      <td>0</td>\n",
       "      <td>1581255876</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1971993</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "      <td>12</td>\n",
       "      <td>4294073</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1581550694</td>\n",
       "      <td>BF36F03D098B39CE46FAC236FD95B880</td>\n",
       "      <td>804689</td>\n",
       "      <td>2437</td>\n",
       "      <td>True</td>\n",
       "      <td>1277762928</td>\n",
       "      <td>13B5FC6FFA16C89544196B61E65DEA51</td>\n",
       "      <td>...</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>407133</td>\n",
       "      <td>54</td>\n",
       "      <td>1</td>\n",
       "      <td>12</td>\n",
       "      <td>3677744</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1581278198</td>\n",
       "      <td>1C024872A6C1EE11014644B375FEEFB8</td>\n",
       "      <td>303</td>\n",
       "      <td>370</td>\n",
       "      <td>False</td>\n",
       "      <td>1498074109</td>\n",
       "      <td>5540A2D31464BA89D5749F31399F5322</td>\n",
       "      <td>...</td>\n",
       "      <td>True</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1581278784</td>\n",
       "      <td>1756945</td>\n",
       "      <td>54</td>\n",
       "      <td>2</td>\n",
       "      <td>6</td>\n",
       "      <td>3582362</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>None</td>\n",
       "      <td>65D3B775D8D7AE83D5D9839C1FE75CDD</td>\n",
       "      <td>D0120C5A771484D1765BBA60461666B9</td>\n",
       "      <td>1580986022</td>\n",
       "      <td>F4DCB14DEA2FB8721C75846B549AFE2B</td>\n",
       "      <td>411495</td>\n",
       "      <td>12</td>\n",
       "      <td>True</td>\n",
       "      <td>1246733300</td>\n",
       "      <td>65834DED849800C03887F514B3885DD4</td>\n",
       "      <td>...</td>\n",
       "      <td>False</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>2092582</td>\n",
       "      <td>38</td>\n",
       "      <td>2</td>\n",
       "      <td>4</td>\n",
       "      <td>2408726</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 24 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "  hashtags                             links  \\\n",
       "0     None                              None   \n",
       "1     None                              None   \n",
       "2     None                              None   \n",
       "3     None                              None   \n",
       "4     None  65D3B775D8D7AE83D5D9839C1FE75CDD   \n",
       "\n",
       "                            domains   timestamp  \\\n",
       "0                              None  1581525142   \n",
       "1                              None  1581252945   \n",
       "2                              None  1581550694   \n",
       "3                              None  1581278198   \n",
       "4  D0120C5A771484D1765BBA60461666B9  1580986022   \n",
       "\n",
       "                          a_user_id  a_follower_count  a_following_count  \\\n",
       "0  46019E3576BD47D19BC83E3A8B04EC35              7717               7688   \n",
       "1  185A653D674A8BDA4C351617C84E5311               781                498   \n",
       "2  BF36F03D098B39CE46FAC236FD95B880            804689               2437   \n",
       "3  1C024872A6C1EE11014644B375FEEFB8               303                370   \n",
       "4  F4DCB14DEA2FB8721C75846B549AFE2B            411495                 12   \n",
       "\n",
       "   a_is_verified  a_account_creation                         b_user_id  ...  \\\n",
       "0          False          1245066162  4992CD23C7A65F9D6CB1F6067D94D09C  ...   \n",
       "1          False          1476364476  5FAA46A494872C5210B49C5CFCFAD331  ...   \n",
       "2           True          1277762928  13B5FC6FFA16C89544196B61E65DEA51  ...   \n",
       "3          False          1498074109  5540A2D31464BA89D5749F31399F5322  ...   \n",
       "4           True          1246733300  65834DED849800C03887F514B3885DD4  ...   \n",
       "\n",
       "   b_follows_a  reply     retweet  retweet_comment        like       id  \\\n",
       "0        False      0           0                0           0  1516532   \n",
       "1         True      0  1581255876                0           0  1971993   \n",
       "2        False      0           0                0           0   407133   \n",
       "3         True      0           0                0  1581278784  1756945   \n",
       "4        False      0           0                0           0  2092582   \n",
       "\n",
       "   language  tweet_type  media  tweet_id  \n",
       "0        54           1     12   2725610  \n",
       "1         0           1     12   4294073  \n",
       "2        54           1     12   3677744  \n",
       "3        54           2      6   3582362  \n",
       "4        38           2      4   2408726  \n",
       "\n",
       "[5 rows x 24 columns]"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "df = df.merge(tweet,on='tweet_id',how='left')\n",
    "df = df.drop('tweet_id',axis=1)\n",
    "df.columns = [i if i!='tweet_encode' else 'tweet_id' for i in df.columns]\n",
    "df, = dask.persist(df)\n",
    "wait(df)\n",
    "del tweet\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "#%%time\n",
    "#df = df.repartition(npartitions=1024)\n",
    "#df, = dask.persist(df)\n",
    "#_ = wait(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "13774339 25315553 146256004\n",
      "30870475 25315553 146256004\n",
      "CPU times: user 5.19 s, sys: 347 ms, total: 5.54 s\n",
      "Wall time: 12.7 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "user_a = df[['a_user_id']].drop_duplicates(split_out=16)\n",
    "user_a, = dask.persist(user_a)\n",
    "_ = wait(user_a)\n",
    "user_b = df[['b_user_id']].drop_duplicates(split_out=16)\n",
    "user_b, = dask.persist(user_b)\n",
    "wait(user_b)\n",
    "print(len(user_a),len(user_b),len(df))\n",
    "\n",
    "user_a.columns = ['user_id']\n",
    "user_b.columns = ['user_id']\n",
    "user_b['dummy'] = 1\n",
    "user_a = user_a.merge(user_b,on='user_id',how='outer')\n",
    "user_a = user_a.drop('dummy',axis=1)\n",
    "user_a, = dask.persist(user_a)\n",
    "wait(user_a)\n",
    "print(len(user_a),len(user_b),len(df))\n",
    "del user_b\n",
    "\n",
    "user_a['user_encode'] = 1\n",
    "user_a['user_encode'] = user_a['user_encode'].cumsum()\n",
    "user_a, = dask.persist(user_a)\n",
    "_ = wait(user_a)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 3.82 s, sys: 275 ms, total: 4.09 s\n",
      "Wall time: 12.4 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df = df.merge(user_a,left_on='a_user_id',right_on='user_id',how='left')\n",
    "df = df.drop(['a_user_id','user_id'],axis=1)\n",
    "df.columns = [i if i!='user_encode' else 'a_user_id' for i in df.columns]\n",
    "df, = dask.persist(df)\n",
    "_ = wait(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 3.78 s, sys: 315 ms, total: 4.1 s\n",
      "Wall time: 12.2 s\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>hashtags</th>\n",
       "      <th>links</th>\n",
       "      <th>domains</th>\n",
       "      <th>timestamp</th>\n",
       "      <th>a_follower_count</th>\n",
       "      <th>a_following_count</th>\n",
       "      <th>a_is_verified</th>\n",
       "      <th>a_account_creation</th>\n",
       "      <th>b_follower_count</th>\n",
       "      <th>b_following_count</th>\n",
       "      <th>...</th>\n",
       "      <th>retweet</th>\n",
       "      <th>retweet_comment</th>\n",
       "      <th>like</th>\n",
       "      <th>id</th>\n",
       "      <th>language</th>\n",
       "      <th>tweet_type</th>\n",
       "      <th>media</th>\n",
       "      <th>tweet_id</th>\n",
       "      <th>a_user_id</th>\n",
       "      <th>b_user_id</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1581167611</td>\n",
       "      <td>38884</td>\n",
       "      <td>145</td>\n",
       "      <td>False</td>\n",
       "      <td>1537006624</td>\n",
       "      <td>12</td>\n",
       "      <td>252</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1581170593</td>\n",
       "      <td>6635832</td>\n",
       "      <td>33</td>\n",
       "      <td>2</td>\n",
       "      <td>8</td>\n",
       "      <td>72414311</td>\n",
       "      <td>1493523</td>\n",
       "      <td>289386</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1581031485</td>\n",
       "      <td>54153</td>\n",
       "      <td>0</td>\n",
       "      <td>False</td>\n",
       "      <td>1564526796</td>\n",
       "      <td>176</td>\n",
       "      <td>265</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1581049466</td>\n",
       "      <td>59662884</td>\n",
       "      <td>47</td>\n",
       "      <td>2</td>\n",
       "      <td>8</td>\n",
       "      <td>37610756</td>\n",
       "      <td>1332557</td>\n",
       "      <td>1176575</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1581115827</td>\n",
       "      <td>476</td>\n",
       "      <td>614</td>\n",
       "      <td>False</td>\n",
       "      <td>1370644514</td>\n",
       "      <td>492</td>\n",
       "      <td>218</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>17001753</td>\n",
       "      <td>3</td>\n",
       "      <td>1</td>\n",
       "      <td>12</td>\n",
       "      <td>72363440</td>\n",
       "      <td>1118690</td>\n",
       "      <td>319351</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1581289083</td>\n",
       "      <td>254</td>\n",
       "      <td>12</td>\n",
       "      <td>False</td>\n",
       "      <td>1507166362</td>\n",
       "      <td>29</td>\n",
       "      <td>105</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>97962265</td>\n",
       "      <td>4</td>\n",
       "      <td>2</td>\n",
       "      <td>4</td>\n",
       "      <td>15680803</td>\n",
       "      <td>1248545</td>\n",
       "      <td>684765</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>None</td>\n",
       "      <td>1581367022</td>\n",
       "      <td>1009</td>\n",
       "      <td>506</td>\n",
       "      <td>False</td>\n",
       "      <td>1240020633</td>\n",
       "      <td>543</td>\n",
       "      <td>450</td>\n",
       "      <td>...</td>\n",
       "      <td>0</td>\n",
       "      <td>0</td>\n",
       "      <td>1581384002</td>\n",
       "      <td>38541399</td>\n",
       "      <td>54</td>\n",
       "      <td>2</td>\n",
       "      <td>12</td>\n",
       "      <td>4639756</td>\n",
       "      <td>1680120</td>\n",
       "      <td>433350</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "<p>5 rows × 24 columns</p>\n",
       "</div>"
      ],
      "text/plain": [
       "  hashtags links domains   timestamp  a_follower_count  a_following_count  \\\n",
       "0     None  None    None  1581167611             38884                145   \n",
       "1     None  None    None  1581031485             54153                  0   \n",
       "2     None  None    None  1581115827               476                614   \n",
       "3     None  None    None  1581289083               254                 12   \n",
       "4     None  None    None  1581367022              1009                506   \n",
       "\n",
       "   a_is_verified  a_account_creation  b_follower_count  b_following_count  \\\n",
       "0          False          1537006624                12                252   \n",
       "1          False          1564526796               176                265   \n",
       "2          False          1370644514               492                218   \n",
       "3          False          1507166362                29                105   \n",
       "4          False          1240020633               543                450   \n",
       "\n",
       "   ...  retweet  retweet_comment        like        id  language  tweet_type  \\\n",
       "0  ...        0                0  1581170593   6635832        33           2   \n",
       "1  ...        0                0  1581049466  59662884        47           2   \n",
       "2  ...        0                0           0  17001753         3           1   \n",
       "3  ...        0                0           0  97962265         4           2   \n",
       "4  ...        0                0  1581384002  38541399        54           2   \n",
       "\n",
       "   media  tweet_id  a_user_id  b_user_id  \n",
       "0      8  72414311    1493523     289386  \n",
       "1      8  37610756    1332557    1176575  \n",
       "2     12  72363440    1118690     319351  \n",
       "3      4  15680803    1248545     684765  \n",
       "4     12   4639756    1680120     433350  \n",
       "\n",
       "[5 rows x 24 columns]"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "df = df.merge(user_a,left_on='b_user_id',right_on='user_id',how='left')\n",
    "df = df.drop(['b_user_id','user_id'],axis=1)\n",
    "df.columns = [i if i!='user_encode' else 'b_user_id' for i in df.columns]\n",
    "df, = dask.persist(df)\n",
    "wait(df)\n",
    "del user_a\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 103 ms, sys: 10.5 ms, total: 114 ms\n",
      "Wall time: 254 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df = df.repartition(npartitions=NP)\n",
    "df, = dask.persist(df)\n",
    "_ = wait(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.52 s, sys: 88.6 ms, total: 1.61 s\n",
      "Wall time: 11.1 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df.to_parquet(f'{path}/dask_input/step1_output',write_index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "This notebook took 2.1 minutes\n"
     ]
    }
   ],
   "source": [
    "print('This notebook took %.1f minutes'%((time.time()-very_start)/60.))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
